package com.lvmama.java.rhino.spark.utils;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SaveMode;

public class SaveToHdfsUtils {
	
	private static final String BASE_DIR = Constants.getInstance().getProperty("hadoop.hdfs.base.dir");
	private static final String HDFS_URL = Constants.getInstance().getProperty("hadoop.hdfs.url");
	private static final SimpleDateFormat format = new SimpleDateFormat("yyyyMMdd");
	public static enum TABLE_NAME {
		API_REQUEST_TIME_DETAIL("请求时间"),
		LOG_DETAIL("日志详情"),
		REQUEST_SPECIAL_PARAM("请求特殊参数"),
		USER_ACCESS_PATH("请求的公共参数");
		private String cnName;
		private TABLE_NAME(String cnName) {
			this.setCnName(cnName);
		}
		public String getCnName() {
			return cnName;
		}
		public void setCnName(String cnName) {
			this.cnName = cnName;
		}
	}
	
	public static <T>void save(SQLContext sqlContext, TABLE_NAME tableName, List<T> dataList) {
		if(dataList == null || dataList.size() == 0) {
			return;
		}
		T data = dataList.get(0);
		DataFrame dataFrame = sqlContext.createDataFrame(dataList, data.getClass());
		save(dataFrame, tableName);
	}
	
	public static <T>void save(SQLContext sqlContext, TABLE_NAME tableName, JavaRDD<T> rdd) {
		T data = rdd.first();
		DataFrame dataFrame = sqlContext.createDataFrame(rdd, data.getClass());
		save(dataFrame, tableName);
	}
	
	private static void save(DataFrame dataFrame, TABLE_NAME tableName){
		String fileName = getFileUri(tableName);
		dataFrame.write().mode(SaveMode.Append).save(fileName);
		System.out.println("数据保存完成：" + fileName);
	}
	
	private static String getFileUri(TABLE_NAME tableName) {
		return getFileUri(tableName, new Date());
	}
	
	public static String getFileUri(TABLE_NAME tableName, Date date) {
		StringBuffer sb = new StringBuffer();
		sb.append(HDFS_URL).append(BASE_DIR);
		sb.append("/").append(format.format(date));
		sb.append("/").append(tableName.name());
		sb.append("/data.parquet");
		return sb.toString();
	}
}
